package org.jetlinks.pro.tenant.impl;

import lombok.extern.slf4j.Slf4j;
import org.hswebframework.ezorm.rdb.mapping.ReactiveRepository;
import org.jetlinks.pro.assets.AssetManager;
import org.jetlinks.pro.assets.AssetType;
import org.jetlinks.pro.tenant.TenantAsset;
import org.jetlinks.pro.tenant.TenantAssetManager;
import org.jetlinks.pro.tenant.event.TenantAssetsBindEvent;
import org.jetlinks.pro.tenant.event.TenantAssetsUnBindAllEvent;
import org.jetlinks.pro.tenant.event.TenantAssetsUnBindEvent;
import org.jetlinks.pro.tenant.impl.entity.AssetMemberBindEntity;
import org.jetlinks.pro.tenant.supports.DefaultTenantAsset;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import javax.annotation.Nonnull;
import javax.annotation.Nullable;
import java.util.*;
import java.util.stream.Collectors;

@Slf4j
@Component
public class DefaultTenantAssetManager implements TenantAssetManager {

    private final AssetManager assetManager;

    private final ReactiveRepository<AssetMemberBindEntity, String> bindRepository;

    private final ApplicationEventPublisher eventPublisher;

    public DefaultTenantAssetManager(@SuppressWarnings("all") ReactiveRepository<AssetMemberBindEntity, String> bindRepository,
                                     ApplicationEventPublisher eventPublisher,
                                     AssetManager assetManager) {
        this.bindRepository = bindRepository;
        this.eventPublisher = eventPublisher;
        this.assetManager = assetManager;
    }

    @Override
    public Mono<AssetType> getAssetType(@Nonnull String typeId) {
        return Mono
            .justOrEmpty(AssetType.of(typeId));
    }

    private TenantAsset convertBindToAsset(AssetMemberBindEntity bind, AssetType assetType) {
        return DefaultTenantAsset.of(bind.getTenantId(), bind.getAssetId(), assetType, bind.getUserId(), bind.getPermission());
    }

    @Override
    public Flux<TenantAsset> getTenantAssets(@Nonnull String tenantId,
                                             @Nonnull AssetType assetType,
                                             @Nullable String userId,
                                             @Nonnull Collection<?> assetId) {
        Assert.notNull(tenantId, "tenantId cannot be null");
        Assert.notNull(assetType, "assetType cannot be null");
        Assert.notNull(assetId, "assetId cannot be null");

        return bindRepository
            .createQuery()
            .where(AssetMemberBindEntity::getAssetType, assetType.getId())
            .and(AssetMemberBindEntity::getTenantId, tenantId)
            .and(AssetMemberBindEntity::getUserId, userId)
            .in(AssetMemberBindEntity::getAssetId, assetId)
            .fetch()
            .map(bind -> convertBindToAsset(bind, assetType));
    }

    @Override
    public Flux<TenantAsset> getTenantAssets(@Nonnull String tenantId,
                                             @Nonnull AssetType assetType,
                                             @Nullable String userId) {
        Assert.notNull(tenantId, "tenantId cannot be null");
        Assert.notNull(assetType, "assetType cannot be null");

        return bindRepository
            .createQuery()
            .where(AssetMemberBindEntity::getAssetType, assetType.getId())
            .and(AssetMemberBindEntity::getTenantId, tenantId)
            .and(AssetMemberBindEntity::getUserId, userId)
            .fetch()
            .map(bind -> convertBindToAsset(bind, assetType));
    }

    @Override
    public Mono<TenantAsset> getTenantAsset(@Nonnull String tenantId,
                                            @Nonnull AssetType assetType,
                                            @Nullable String userId,
                                            @Nonnull String assetId) {
        return this
            .getTenantAssets(tenantId, assetType, userId, Collections.singletonList(assetId))
            .singleOrEmpty();
    }

    @Override
    public Mono<Void> bindAssetsOwner(@Nonnull String tenantId,
                                      @Nonnull AssetType assetType,
                                      @Nonnull String userId,
                                      @Nonnull Collection<?> assetId) {
        Assert.notNull(tenantId, "tenantId cannot be null");
        Assert.notNull(assetType, "assetType cannot be null");
        Assert.notNull(userId, "userId cannot be null");
        Assert.notNull(assetId, "assetId cannot be null");

        return bindAssets(tenantId,
                          false,
                          Flux.just(BindAssetsRequest
                                        .create()
                                        .assetType(assetType.getId())
                                        .assetIdList(assetId
                                                         .stream()
                                                         .map(String::valueOf)
                                                         .collect(Collectors.toList()))
                                        .allPermission(true)
                                        .userId(userId)
                                        .build()))
            .then();
    }

    @Transactional
    public Mono<Integer> unbindAssets(@Nonnull String tenantId,
                                      @Nonnull Flux<UnbindAssetsRequest> requestStream) {
        Assert.notNull(tenantId, "tenantId cannot be null");
        Assert.notNull(requestStream, "requestStream cannot be null");

        return requestStream
            .doOnNext(UnbindAssetsRequest::validate)
            .flatMap(request -> bindRepository
                .createDelete()
                .where(AssetMemberBindEntity::getTenantId, tenantId)
                .and(AssetMemberBindEntity::getUserId, request.getUserId())
                .and(AssetMemberBindEntity::getAssetType, request.getAssetType())
                .in(AssetMemberBindEntity::getAssetId, request.getAssetIdList())
                .execute()
                //推送解绑事件
                .flatMap(TenantAssetsUnBindEvent
                             .of(tenantId, request.getAssetType(), request.getUserId(), request.getAssetIdList())
                             .publish(eventPublisher)
                             ::thenReturn)
            )
            .reduce(Math::addExact);
    }

    @Override
    public Mono<Integer> unbindAllAssets(@Nonnull String assetType, @Nonnull Collection<?> assertIds) {
        Assert.notNull(assetType, "assetType cannot be null");
        Assert.notNull(assertIds, "assertIds cannot be null");

        return bindRepository
            .createDelete()
            .where(AssetMemberBindEntity::getAssetType, assetType)
            .in(AssetMemberBindEntity::getAssetId, assertIds)
            .execute()
            .flatMap(TenantAssetsUnBindAllEvent
                         .of(assetType, assertIds.stream().map(String::valueOf).collect(Collectors.toSet()))
                         .publish(eventPublisher)
                         ::thenReturn);
    }

    @Transactional
    public Mono<Integer> bindAssets(@Nonnull String tenantId,
                                    boolean filterOutsideAsset,
                                    @Nonnull Flux<BindAssetsRequest> requestStream) {
        Assert.notNull(tenantId, "tenantId cannot be null");
        Assert.notNull(requestStream, "requestStream cannot be null");

        return requestStream
            .flatMap(request -> request.toBindEntity(tenantId, this))
            .distinct(AssetMemberBindEntity::getId)
            .buffer(200)
            .flatMap(list -> {
                if (!filterOutsideAsset) {
                    return this.doBind(list);
                }
                return Flux
                    .fromIterable(list)
                    .collect(Collectors.groupingBy(
                        AssetMemberBindEntity::getAssetType,
                        Collectors.mapping(AssetMemberBindEntity::getAssetId, Collectors.toList())))
                    .flatMapIterable(Map::entrySet)
                    .flatMap(entry -> bindRepository
                        .createQuery()
                        .where(AssetMemberBindEntity::getTenantId, tenantId)
                        .and(AssetMemberBindEntity::getAssetType, entry.getKey())
                        .in(AssetMemberBindEntity::getAssetId, entry.getValue())
                        .fetch()
                        .map(bind -> bind.getAssetType() + ":" + bind.getAssetId())
                        .collectList())
                    .flatMap(insideAsset -> this
                        .doBind(list.stream()
                                    .filter(bind -> {
                                        if (!insideAsset.contains(bind.getAssetType() + ":" + bind.getAssetId())) {
                                            log.warn("租户[{}]没有资产权限{}:{}", tenantId, bind.getAssetType(), bind.getAssetId());
                                            return false;
                                        }
                                        return true;
                                    })
                                    .collect(Collectors.toList())));
            })
            .reduce(Math::addExact);

    }

    protected Mono<Integer> doBind(List<AssetMemberBindEntity> binds) {
        if (CollectionUtils.isEmpty(binds)) {
            return Mono.just(0);
        }
        return bindRepository
            .save(binds)
            .flatMap(i -> Flux
                .fromIterable(binds)
                .groupBy(bind -> String.join(
                    ":", bind.getTenantId(), bind.getAssetType(), bind.getUserId()
                ))
                //推送资产绑定事件
                .flatMap(group -> group
                    .switchOnFirst((signal, flux) -> {
                        AssetMemberBindEntity first = signal.get();
                        if (first == null) {
                            return flux.then();
                        }
                        return flux
                            .map(AssetMemberBindEntity::getAssetId)
                            .collectList()
                            .flatMap(assetsId -> TenantAssetsBindEvent
                                .of(
                                    first.getTenantId(),
                                    first.getAssetType(),
                                    first.getUserId(),
                                    assetsId
                                )
                                .publish(eventPublisher));

                    }))
                .then(Mono.just(i.getTotal()))
            );
    }

    @Override
    @Transactional
    public Mono<Void> unbindAssetsOwner(@Nonnull String tenantId,
                                        @Nonnull AssetType assetType,
                                        @Nonnull String userId,
                                        @Nonnull Collection<?> assetId) {
        return unbindAssets(tenantId,
                            Flux.just(UnbindAssetsRequest
                                          .builder()
                                          .assetType(assetType.getId())
                                          .assetIdList(assetId
                                                           .stream()
                                                           .map(String::valueOf)
                                                           .collect(Collectors.toList()))
                                          .userId(userId)
                                          .build()))
            .then();
    }

    @Override
    @Transactional
    public Mono<Void> unbindAssetsOwner(@Nonnull AssetType assetType,
                                        @Nonnull Collection<String> userIdList,
                                        @Nonnull Collection<?> assetId) {

        return bindRepository
            .createQuery()
            .where(AssetMemberBindEntity::getAssetType, assetType.getId())
            .in(AssetMemberBindEntity::getUserId, userIdList)
            .in(AssetMemberBindEntity::getAssetId, assetId)
            .fetch()
            .groupBy(AssetMemberBindEntity::getTenantId)
            .flatMap(tenantGroup -> {
                String tenantId = String.valueOf(tenantGroup.key());
                return bindRepository
                    .createDelete()
                    .where(AssetMemberBindEntity::getTenantId, tenantId)
                    .in(AssetMemberBindEntity::getUserId, userIdList)
                    .and(AssetMemberBindEntity::getAssetType, assetType.getId())
                    .in(AssetMemberBindEntity::getAssetId, assetId)
                    .execute()
                    .then(
                        tenantGroup
                            .groupBy(AssetMemberBindEntity::getUserId)
                            .flatMap(userIdGroup -> {
                                String userId = userIdGroup.key();
                                return userIdGroup
                                    .map(AssetMemberBindEntity::getAssetId)
                                    .collectList()
                                    .flatMap(assetIds -> TenantAssetsUnBindEvent
                                        .of(tenantId, assetType.getId(), userId, assetIds)
                                        .publish(eventPublisher));
                            })
                            .then()
                    );
            })
            .then();
    }

    @Override
    public Mono<org.jetlinks.pro.assets.Asset> getAsset(@Nonnull AssetType assetType,
                                                        @Nonnull String assetId) {
        return assetManager.getAsset(assetType, assetId);
    }

    @Override
    public Flux<org.jetlinks.pro.assets.Asset> getAssets(@Nonnull AssetType assetType,
                                                         @Nonnull Collection<?> assetId) {
        return assetManager.getAssets(assetType, assetId);
    }

    @Override
    public Flux<TenantAsset> getTenantAssets(@Nonnull AssetType assetType,
                                             @Nonnull Collection<?> assetId) {
        if (CollectionUtils.isEmpty(assetId)) {
            return Flux.empty();
        }
        return bindRepository
            .createQuery()
            .where(AssetMemberBindEntity::getAssetType, assetType.getId())
            .in(AssetMemberBindEntity::getAssetId, assetId)
            .fetch()
            .map(bind -> convertBindToAsset(bind, assetType));
    }

    @Override
    public Flux<TenantAsset> getTenantAssets(@Nonnull AssetType assetType,
                                             @Nonnull String userId,
                                             @Nonnull Collection<?> assetId) {
        if (CollectionUtils.isEmpty(assetId)) {
            return Flux.empty();
        }
        Objects.requireNonNull(assetType, "assetType can not be null");
        Objects.requireNonNull(userId, "userId can not be null");
        return bindRepository
            .createQuery()
            .where(AssetMemberBindEntity::getAssetType, assetType.getId())
            .and(AssetMemberBindEntity::getUserId, userId)
            .in(AssetMemberBindEntity::getAssetId, assetId)
            .fetch()
            .map(bind -> convertBindToAsset(bind, assetType));
    }

}
